package com.example.hzq;

import com.lmax.disruptor.RingBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Netty处理器，将消息发布到Disruptor
 */
public class DisruptorHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(DisruptorHandler.class);
    private final RingBuffer<MessageEvent> ringBuffer;
    
    public DisruptorHandler(RingBuffer<MessageEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            // 从Netty获取消息
            ByteBuf buf = (ByteBuf) msg;
            byte[] bytes = new byte[buf.readableBytes()];
            buf.readBytes(bytes);
            String message = new String(bytes);
            
            // 发布到Disruptor
            long sequence = ringBuffer.next();
            try {
                MessageEvent event = ringBuffer.get(sequence);
                event.setMessage(message);
                event.setCtx(ctx);
            } finally {
                ringBuffer.publish(sequence);
            }
        } finally {
            // 释放资源
            ((ByteBuf) msg).release();
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("Exception caught", cause);
        ctx.close();
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        logger.info("New connection established: {}", ctx.channel().remoteAddress());
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        logger.info("Connection closed: {}", ctx.channel().remoteAddress());
    }
}
